package com.zhao.biz.report

import com.typesafe.config.Config
import com.zhao.entity.Log
import com.zhao.utils.CommonUtil
import org.apache.spark.SparkContext
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

import java.util.Properties

/**
 * Description: 报表展示之共通处理spark sql版本<br/>
 * Copyright (c) ，2021 ， 赵 <br/>
 * A wet person does not fear the rain. <br/>
 * Date： 2021/1/13 14:57
 *
 * @author 柒柒
 * @version : 1.0
 */

object CommonCal {

  def commonDealWith(args: Array[String],groupField: String, targetTable: String) = {
    //步骤:
    val Array(input) = args

    //1.SparkSession
    val spark: SparkSession = SparkSession
      .builder
      .appName(this.getClass.getSimpleName)
      .master("local[*]")
      //设置序列化的技术(使用kryo)
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .getOrCreate()

    val sc: SparkContext = spark.sparkContext

    //2.注册实体类(本质:哪个实体类的实例使用了kryo序列化技术)
    sc.getConf.registerKryoClasses(Array(classOf[Log]))

    //3.加载ETL后数据到内存中,装载为DataFrame,并映射为一张虚拟表
    spark.read.parquet(input).createOrReplaceTempView("tb_log")

    spark.sqlContext.cacheTable("tb_log")

    //4.使用spark sql统计
    val resultDF: DataFrame = spark.sql(
      s"""
         |select
         | $groupField,
         | originalReq,
         | effectiveReq,
         | adReq,
         | bidCnt,
         | bidSuccCnt,
         | concat(cast(if(bidSuccCnt=0,0,bidSuccCnt*100.0/bidCnt) as decimal(10,2)),'%')  bidSuccRate,
         | showCnt,
         | clickCnt,
         | if(clickCnt=0,0,clickCnt*1.0/showCnt) clickRate,
         | adCost,
         | adConsumer
         |from(
         |   select
         |    $groupField,
         |    sum(case when requestmode=1 and processnode>=1 then 1 else 0 end ) originalReq, -- 原始请求数
         |    sum(case when requestmode=1 and processnode>=2 then 1  else 0 end) effectiveReq, -- 有效请求数
         |    sum(case when requestmode=1 and processnode=3  then 1  else 0 end) adReq, -- 广告请求数
         |    sum(case when iseffective=1 and isbilling=1 and isbid=1 then 1  else 0 end) bidCnt, -- 参与竞价数
         |    sum(case when iseffective=1 and isbilling=1 and iswin=1 and adorderid!=0 then 1  else 0 end) bidSuccCnt, -- 竞价成功数
         |    sum(case when requestmode=2 and  iseffective=1   then 1  else 0 end) showCnt, -- 展示量
         |    sum(case when requestmode=3 and  iseffective=1   then 1  else 0 end) clickCnt, -- 点击量
         |    sum(case when iseffective=1  and isbilling=1  and  iswin=1 then adpayment/1000 else 0 end)    adCost, -- 广告成本（广告公司支付给网站主（媒体））
         |    sum(case when iseffective=1  and isbilling=1  and  iswin=1 then winprice/1000 else 0 end)   adConsumer -- 广告消费 （广告主给广告公司）
         |   from tb_log
         |group by $groupField) t
      """.stripMargin)

    //展示测试
    resultDF.show(10000)

    //5.落地
    //落地到DB中
    //保存到mysql中
//    val load: Config = CommonUtil.load
//    //设置请求mysql的配置信息
//    val prop: Properties = new Properties()
//    prop.setProperty("user",load.getString("db.default.user"))
//    prop.setProperty("password",load.getString("db.default.password"))
//    resultDF.write.mode(SaveMode.Overwrite)
//      .jdbc(load.getString("db.default.url"),targetTable,prop)

    //6.资源释放
    spark.stop()
  }
}
